package com.wpx.batch.reader;

import com.wpx.batch.ExecuteContext;
import com.wpx.batch.ItemReader;
import com.wpx.batch.WpxBatchUtils;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 接收读
 */
public class ReceptorItemReader<E> implements ItemReader<E> {

    private Queue<E> bufferQueue = new ConcurrentLinkedQueue<>();

    @Override
    public List<E> read(ExecuteContext executeContext) {
        while (true){
            if(executeContext.getJob().isWillStop()){
                return new ArrayList<>(); //作业将结束 返回空集合
            }
            if(this.bufferQueue.size() > 0){
                List<E> data = new ArrayList<>();
                while(true){
                    E e = this.bufferQueue.poll();
                    if(e == null)break;
                    data.add(e);
                }
                return data;
            }else{
                WpxBatchUtils.sleep(500);  //一直没数据 等待500毫秒 再去读
            }
        }
    }

    public void receptor(E item){
        if(item == null) return;
        this.bufferQueue.add(item);
    }

    public void receptor(List<E> items){
        if(WpxBatchUtils.isEmpty(items)) return;
        for (E item : items) {
            this.receptor(item);
        }
    }
}
